package com.guass.nettyprotostuff.zookeeper;

import com.alibaba.fastjson.JSON;
import com.guass.nettyprotostuff.conf.ServerProperties;
import com.guass.nettyprotostuff.conf.ZKProperties;
import com.guass.nettyprotostuff.rpc.client.ClientTest;
import com.guass.nettyprotostuff.service.RpcService;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.data.Stat;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.stream.Collectors;

/**
 * @author guass
 * @date 2022年03月09日 14:04
 */
@Slf4j
@Service
public class ZookeeperService {


    private CuratorFramework curatorFramework;

    @Resource
    private ZKProperties zkProperties;

    @Resource
    private ServerProperties serverProperties;

    @Value("${spring.application.name}")
    private String nodeName;

    @Resource
    private RpcService rpcService;

    private String forPath;

    private DataInfo serverInfo;

    public String getForPath() {
        return forPath;
    }

    public DataInfo getServerInfo() {
        return serverInfo;
    }

    public CuratorFramework getCuratorFramework() {
        return curatorFramework;
    }

    public String getData(String path) {
        try {
            if (curatorFramework != null) {
                return new String(curatorFramework.getData().forPath(path));
            }

        }catch(Exception e){
            e.printStackTrace();
        }
        return null;
    }

    public String getServerIp() {
        try {
            List<String> pathlist = listAll("/node");

            if (pathlist != null && pathlist.size() > 0) {

                List<DataInfo> dataInfoList = new ArrayList<>();

                for (String path : pathlist) {
                    String data = getData("/node" +"/" +path);

                    DataInfo dataInfo = JSON.parseObject(data, DataInfo.class);
                    log.info("guass data {}", dataInfo);
                    dataInfoList.add(dataInfo);
                }

                dataInfoList = dataInfoList.stream().sorted(Comparator.comparingInt(DataInfo::getCounter)).collect(Collectors.toList());


                return dataInfoList.get(0).getIp();
            }


        }catch (Exception e){
            e.printStackTrace();
        }
        return null;
    }

    public  List<String> listAll(String path) throws Exception {
        List<String> strings = curatorFramework.getChildren().forPath(path);
        return  strings;
    }

    public void start(ZkConnect zkConnect){

        if (!StringUtils.isEmpty(zkProperties.getConnectString()) && serverProperties.isUseZkConnect()) {

            RetryPolicy retryPolicy = new ExponentialBackoffRetry(zkProperties.getBaseSleepTimeMs(), zkProperties.getMaxRetries());
            curatorFramework = CuratorFrameworkFactory.builder().connectString(zkProperties.getConnectString())
                    .retryPolicy(retryPolicy)
                    .sessionTimeoutMs(zkProperties.getSession_timeout_ms())
                    .connectionTimeoutMs(zkProperties.getConnection_timeout_ms())
                    .canBeReadOnly(true)
                    .build();
            curatorFramework.getConnectionStateListenable().addListener((client, newState) -> {
                if (newState == ConnectionState.CONNECTED) {
                    log.info("guass zk 连接成功！");
                    zkConnect.onConnected();
                    try {
                        addToZk();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }

            });
            log.info("guass zk 连接中......");
            curatorFramework.start();
        }
    }


    public void updateCounter(int channelCount){

        if (serverProperties.isUseZkConnect()) {
            if (serverInfo != null) {
                serverInfo.setCounter(channelCount);

                try {
                    curatorFramework.setData().forPath(forPath, JSON.toJSONString(serverInfo).getBytes(StandardCharsets.UTF_8));
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    public void addToZk() throws Exception {

        if (serverProperties.isUseZkConnect()) {

            String path = zkProperties.getPath();

            serverInfo = new DataInfo();

            serverInfo.setCounter(rpcService.getChannelCount());
            serverInfo.setIp(serverProperties.getIp());

            forPath = curatorFramework
                    .create()
                    .creatingParentContainersIfNeeded()
//                .creatingParentsIfNeeded()
                    .withProtection()
                    .withMode(CreateMode.EPHEMERAL_SEQUENTIAL).
                    forPath(path, JSON.toJSONString(serverInfo).getBytes(StandardCharsets.UTF_8));
            log.info("guass create node :{} {} successfully.", forPath,serverInfo);
        }


    }


    private void createIfNeed(String path) throws Exception {

        if (serverProperties.isUseZkConnect()) {
            Stat stat = curatorFramework.checkExists().forPath(path);
            if (stat==null){
                String s = curatorFramework.create().forPath(path);
                log.info("path {} created! ",s);
            }
        }
    }
}
